package io.jnet.common.businessprocessor;

import java.util.concurrent.BlockingQueue;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import io.jnet.common.api.ChannelContext;
import io.jnet.common.streamprocessor.ProcesserUtil;
import io.jnet.common.streamprocessor.ProcessorTask;
import io.jnet.common.util.AioListener;

public class ConcentrationProcessor implements Runnable
{
    private final BlockingQueue<ProcessorTask> tasks;
    private final AioListener                  listener;
    
    public ConcentrationProcessor(BlockingQueue<ProcessorTask> tasks, AioListener listener)
    {
        this.tasks = tasks;
        this.listener = listener;
    }
    
    @Override
    public void run()
    {
        ProcessorTask task = null;
        while (true)
        {
            try
            {
                task = tasks.take();
                ChannelContext channelContext = task.getChannelContext();
                channelContext.processors();
                Object finalResult = ProcesserUtil.process(channelContext, channelContext.processors(), task.getData(), task.getInitIndex());
                if (finalResult instanceof ByteBuf<?>)
                {
                    channelContext.bufStorage().putBuf((ByteBuf<?>) finalResult);
                    channelContext.registerWrite();
                }
            }
            catch (Throwable e)
            {
                if (e instanceof InterruptedException)
                {
                    return;
                }
                listener.catchException(e, task.getChannelContext());
            }
        }
    }
    
}
